package com.leilei;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/11 20:32
 * @desc 数据分流 1.12之前是 split 和select  现在为 Side Outputs
 */
public class SideOutPutOperator {

    public static final String JON = "jon";
    public static final String LUCY = "lucy";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(1);
        // 原始流
        DataStreamSource<String> ds = env.fromElements("jon.a", "jon.b", "lucy.zs", "lucy.ls");
        // 定义拆分后的侧面输出标识 （名字 和元素类型）
        OutputTag<String> jonTag = new OutputTag<>(JON, TypeInformation.of(String.class));
        // 定义拆分后的侧面输出标识 （名字 和元素类型）
        OutputTag<String> lucyTag = new OutputTag<>(LUCY, TypeInformation.of(String.class));
        //将一个流切分为多个流
        SingleOutputStreamOperator<String> source = ds.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) {
                // 自定义数据归属判断 ex:以jon 打头数据则输出到jonTag ；以lucy打头数据则输出到lucyTag
                if (value.startsWith(JON)) {
                    ctx.output(jonTag , value);
                }
                if (value.startsWith(LUCY)) {
                    ctx.output(lucyTag, value);
                }
            }
        });
        // 获取侧面输出数据 传入侧面输出标识.即可从原始流数据源中获取拆分后的数据,形成一个全新的数据流
        DataStream<String> jonStream = source.getSideOutput(jonTag);
        DataStream<String> lucyStream = source.getSideOutput(lucyTag);
        jonStream.print(JON);
        lucyStream.print(LUCY);
        env.execute("side-process");

    }
}
